package com.allscore.storm.demo.kafka;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;


public class PrintBolt extends BaseBasicBolt {

    private static final long serialVersionUID = 1L;

    public static final Log log = LogFactory.getLog(PrintBolt.class);

    private FileWriter fileWriter;

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        //获取上一个组件所声明的Field
        try {
            String print = input.getString(0);
            log.info(print);
            fileWriter.write(print + "\r\n");
            fileWriter.flush();

        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // Do nothing because end.
    }


    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        try {
            fileWriter = new FileWriter("D:/tmp/test.log");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

}
